package com.atguigu.day06;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import com.atguigu.util.UserViewCountPerWindow;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Example4 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ClickEvent>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((element, recordTimestamp) -> element.ts)
                )
                .keyBy(r -> r.username)
                .window(TumblingEventTimeWindows.of(Time.days(1)))
                .trigger(new Trigger<>() {
                    @Override
                    public TriggerResult onElement(ClickEvent clickEvent, long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        // 为第一条数据的事件时间后面的整数秒注册一个定时器
                        // 窗口状态变量，每个窗口都维护自己的窗口状态变量
                        // 标志位
                        ValueState<Boolean> flag = triggerContext.getPartitionedState(
                                new ValueStateDescriptor<>(
                                        "flag",
                                        Types.BOOLEAN
                                )
                        );

                        if (flag.value() == null) {
                            // 计算接下来的整数秒
                            // 1234ms ===> 2000ms
                            long nextSecond = clickEvent.ts + 1000L - clickEvent.ts % 1000L;
                            // 注册事件时间定时器，也就是onEventTime
                            triggerContext.registerEventTimeTimer(nextSecond);

                            flag.update(true);
                        }

                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) {
                        return null;
                    }

                    @Override
                    public TriggerResult onEventTime(long timerTs, TimeWindow w, TriggerContext ctx) {
                        if (timerTs < w.getEnd() - 1L) {
                            if (timerTs + 1000L < w.getEnd() - 1L) {
                                // 注册下一个整数秒的定时器
                                ctx.registerEventTimeTimer(timerTs + 1000L);
                            }
                            // 触发窗口计算，但不销毁
                            return TriggerResult.FIRE;
                        }

                        // flink注册的默认的定时器：窗口结束时间 - 1毫秒
                        return TriggerResult.FIRE_AND_PURGE;
                    }

                    @Override
                    public void clear(TimeWindow timeWindow, TriggerContext triggerContext) {
                        // 单例特性
                        ValueState<Boolean> flag = triggerContext.getPartitionedState(
                                new ValueStateDescriptor<>(
                                        "flag",
                                        Types.BOOLEAN
                                )
                        );

                        flag.clear();
                    }
                })
                .aggregate(
                        new AggregateFunction<ClickEvent, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(ClickEvent value, Long accumulator) {
                                return accumulator + 1L;
                            }

                            @Override
                            public Long getResult(Long accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return null;
                            }
                        },
                        new ProcessWindowFunction<Long, UserViewCountPerWindow, String, TimeWindow>() {
                            @Override
                            public void process(String key, Context context, Iterable<Long> iterable, Collector<UserViewCountPerWindow> collector) {
                                collector.collect(new UserViewCountPerWindow(
                                        key,
                                        iterable.iterator().next().intValue(),
                                        context.window().getStart(),
                                        context.window().getEnd()
                                ));
                            }
                        }
                )
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
